package com.future.WebSocket;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.future.allUtils.constants.RedisCacheConstants;
import com.future.domain.ChatMessage;
import com.future.domain.DTO.MessageDTO;
import com.future.domain.DTO.NotificationDTO;
import com.future.domain.User;
import com.future.mapper.ChatRoomMemberMapper;
import com.future.mapper.UserMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.*;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Component
@Slf4j
public class WebSocketSessionManager {
    @Autowired
    private ChatRoomMemberMapper chatRoomMemberMapper;
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    // 注入线程池
    @Autowired
    @Qualifier("chatMessagePool")
    private ExecutorService chatMessagePool;
    private final Map<Long, WebSocketSession> userSessionMap = new ConcurrentHashMap<>();

    /**
     * 存储用户会话
     *
     * @param userId
     * @param session
     */
    public void addSession(Long userId, WebSocketSession session) {
        if (userId == null || session == null) {
            return;
        }
        // 关闭旧会话（避免重连导致的会话堆积）
        WebSocketSession oldSession = userSessionMap.get(userId);
        if (oldSession != null && oldSession.isOpen()) {
            try {
                oldSession.close(CloseStatus.NORMAL.withReason("用户重连，关闭旧会话"));
                log.info("用户重连，关闭旧会话，userId:{}", userId);
            } catch (IOException e) {
                log.error("关闭旧会话失败，userId:{}", userId, e);
            }
        }
        userSessionMap.put(userId, session);
    }

    /**
     * 移除用户会话
     */
    public void removeSession(Long userId) {
        WebSocketSession session = userSessionMap.remove(userId);
        if (session != null && session.isOpen()) {
            try {
                session.close(CloseStatus.NORMAL.withReason("用户主动离线"));
                log.info("移除并关闭会话，userId:{}", userId);
            } catch (IOException e) {
                log.error("关闭会话失败，userId:{}", userId, e);
            }
        }
    }

    /**
     * 根据用户id获取用户会话
     *
     * @return
     */
    public WebSocketSession getSession(Long userId) {
        return userSessionMap.get(userId);

    }

    /**
     * 向指定用户发送消息
     */
    public void sendMessageToUser(Long userId, MessageDTO messageDTO) throws IOException {
        WebSocketSession session = getSession(userId);
        if (session != null && session.isOpen()) {
            String textMessage = objectMapper.writeValueAsString(messageDTO);
            session.sendMessage(new TextMessage(textMessage));
        } else {
            log.info("用户离线，消息将存储为离线消息，userId:{}", userId);
        }
    }

    /**
     * 广播消息到聊天室所有成员
     *
     * @param roomId     聊天室ID
     * @param messageDTO 消息内容
     */
    public void broadcastToRoom(Long roomId, MessageDTO messageDTO) {
        try{
        //获取聊天室成员
            String memberKey = RedisCacheConstants.CHAT_ROOM_MEMBERS_PREFIX + roomId;
            Set<String> members = stringRedisTemplate.opsForSet().members(memberKey);
            if(members==null)
            {
                members=Collections.emptySet();
            }
            List<Long> userIds= members.stream()
                    .filter(Objects::nonNull)  // 过滤null
                    .map(Long::valueOf)
                    .collect(Collectors.toList());
            //如果redis未命中，查询数据库
            if(userIds.isEmpty())
            {
                userIds= chatRoomMemberMapper.selectUserIdByRoomId(roomId);
                if (userIds == null) {
                    return;
                }
                //存入redis
                String[] array = userIds.stream()
                        .map(String::valueOf)
                        .toArray(String[]::new);
                stringRedisTemplate.opsForSet().add(memberKey,array);
                stringRedisTemplate.expire(memberKey, RedisCacheConstants.CACHE_NORMAL_TTL, TimeUnit.SECONDS);
            }
            log.info("聊天室 {} 的成员列表: {}", roomId, members);
        for (Long memberId : userIds) {
            //排除发送者自己
            if (memberId.equals(Long.valueOf(messageDTO.getSenderId()))) {
                continue;
            }
                Runnable runnable = () -> {
                    try {
                        sendMessageToUser(memberId, messageDTO);
                    } catch (IOException e) {
                        log.error("向用户发送消息失败，userId:{}", memberId, e);
                    }
                };
                //提交任务到线程池异步执行
                chatMessagePool.execute(runnable);
            }}catch (Exception e)
        {
            log.error("消息发送失败"+e);
        }
        }
}
